package com.atguigu.app

import java.sql.{Connection, PreparedStatement, ResultSet}
import java.text.SimpleDateFormat
import java.{sql, util}
import java.util.Date

import com.alibaba.fastjson.JSON
import com.alibaba.fastjson.serializer.SerializeConfig
import com.atguigu.bean.{OrderDetail, OrderInfor2, SaleDetail, UserInfo}
import com.atguigu.common.logger.GmallConstants
import com.atguigu.utils.{EsUtils, JDBCUtils, MykafkaUtil, RedisUtil}
import org.apache.spark.SparkConf
import org.apache.spark.streaming._
import redis.clients.jedis.Jedis

import scala.collection.mutable.ListBuffer

/**
 * 需求：mysql实时查询订单详情
 * 订单表(order_detail) + 订单详情(order_info) + 用户表(user_info)
 * 使用双流join
 *
 * @author WangJX
 * @date 2019/12/2 21:23 
 * @version 1.0
 */

object SaleApp {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setAppName("SaleApp")
      .set("spark.streaming.kafka.maxRatePerPartition", "100")
      .set("spark.streaming.backpressure.enabled", "true")
      .set("spark.streaming.stopGracefullyOnShutdown", "true")
      //.setMaster("local[*]")

    val ssc = new StreamingContext(conf, Seconds(5))

    //读取msyql中order_detail表中的数据
    val detailDstream = MykafkaUtil.getKafkaStream(GmallConstants.KAFKA_TOPIC_ORDER_DETAIL, ssc)
      .map(_.value())
      //.window(Seconds(600), Seconds(5))
      .map {
        rdd => {
          val detailInfo: OrderDetail = JSON.parseObject(rdd, classOf[OrderDetail])
          (detailInfo.order_id, detailInfo)
        }
      }

    //读取mysql中order_info表中的数据
    val orderDstream = MykafkaUtil.getKafkaStream(GmallConstants.KAFKA_TOPIC_NEW_ORDER, ssc)
      .map(_.value())
      //      .window(Seconds(600), Seconds(5))
      .map {
        rdd => {
          //对数据进行脱敏
          val orderInfo: OrderInfor2 = JSON.parseObject(rdd, classOf[OrderInfor2])

          //对电话号码进行脱敏
          val phone: String = orderInfo.consignee_tel
          val frePhone: String = phone.substring(0, 3)
          val fixPhone: String = phone.substring(phone.length - 4, phone.length)
          val newPhone = frePhone + "****" + fixPhone
          orderInfo.consignee_tel = newPhone

          //完善采集时间
          val dateString: String = new SimpleDateFormat("yyyy-MM-dd HH").format(new Date)
          val date: Array[String] = dateString.split(" ")
          orderInfo.create_date = date(0)
          orderInfo.create_hour = date(1)

          (orderInfo.id, orderInfo)
        }
      }


    //121行使用scala中的for循环时导入
    import scala.collection.JavaConversions._
    //使用双流full合并
    val mergeDstream = detailDstream
      .fullOuterJoin(orderDstream)
      .flatMap {
        case (id, (detail, order)) => {

          val jedis: Jedis = RedisUtil.getJedisPool()

          //输出集合
          val list = ListBuffer[SaleDetail]()


          //匹配成功，放入输出队列
          if (order != None && detail != None) {
            //匹配成功的数据
            val saleDetail = new SaleDetail(order.get, detail.get)
            list += saleDetail

            //可能后面还有数据未进行匹配，所以需要把数据都存入redis
            val orderInfo: OrderInfor2 = order.get
            val orderInfoStr: String = JSON.toJSONString(orderInfo, new SerializeConfig(true))
            val orderKey = "order:" + orderInfo.id
            //保存数据到redis中(并设置到期时间为10分钟)       string结构
            jedis.setex(orderKey, 600, orderInfoStr)


            //可能detail订单来早了，保存进redis中，设置保存时间为10分钟
            val orderDetail: OrderDetail = detail.get
            val orderDetailStr: String = JSON.toJSONString(orderDetail, new SerializeConfig(true))
            val detailKey = "detail:" + orderDetail.order_id
            //保存数据到redis中 并设置到期时间             保存到set集合中
            jedis.sadd(detailKey, orderDetailStr)
            jedis.expire(detailKey, 600)

          } else if (order != None && detail == None) {
            //可能order_info来早了，保存进redis中，设置保存时间为10分钟
            val orderInfo: OrderInfor2 = order.get
            val orderInfoStr: String = JSON.toJSONString(orderInfo, new SerializeConfig(true))
            val key = "order:" + orderInfo.id
            //保存数据到redis中(并设置到期时间为10分钟)       string结构
            jedis.setex(key, 600, orderInfoStr)


            //读取redis中的detail订单
            val redisDetail: util.Set[String] = jedis.smembers("detail:" + orderInfo.id)
            if (redisDetail != null && redisDetail.size() > 0) {
              for (elem <- redisDetail) {
                val detStr: OrderDetail = JSON.parseObject(elem, classOf[OrderDetail])
                val saleDetail = new SaleDetail(orderInfo, detStr)
                list += saleDetail
              }
            }

          } else if (detail != None && order == None) {
            //可能detail订单来早了，保存进redis中，设置保存时间为10分钟
            val orderDetail: OrderDetail = detail.get
            val orderDetailStr: String = JSON.toJSONString(orderDetail, new SerializeConfig(true))
            val key = "detail:" + orderDetail.order_id
            //保存数据到redis中 并设置到期时间             保存到set集合中
            jedis.sadd(key, orderDetailStr)
            jedis.expire(key, 600)


            //读取redis中的orderInfo订单
            val redisDetail = jedis.get("order:" + orderDetail.order_id)
            if (redisDetail != null && redisDetail.length > 0) {
              val order: OrderInfor2 = JSON.parseObject(redisDetail, classOf[OrderInfor2])
              val saleDetail = new SaleDetail(order, orderDetail)
              list += saleDetail
            }
          }

          jedis.close()
          list
        }
      }
/*          .foreachRDD {
            rdd => {
              rdd.foreach(println)
            }
          }*/


    //合并user_info数据表
    val mergeUserInfoDstream = mergeDstream.mapPartitions {
      rdd => {

        val list = ListBuffer[(String, SaleDetail)]()

        //redis连接
        val jedis: Jedis = RedisUtil.getJedisPool()

        //mysql连接
        val conn: Connection = JDBCUtils.getConnection
        val statement: PreparedStatement = conn.prepareStatement("select * from user_info where id = ?")

        //获取redis中user_info中的所有key的值
        val ids: util.Set[String] = jedis.hkeys(GmallConstants.REDIS_USER_INFO)

        for (elem <- rdd) {
          //如果redis中有该用户修改后的信息
          if (ids != null && ids.contains(elem.user_id)) {
            //获取redis中的用户信息
            val value: String = jedis.hget(GmallConstants.REDIS_USER_INFO, elem.user_id)
            val userInfo: UserInfo = JSON.parseObject(value, classOf[UserInfo])
            //合并user_info信息
            elem.mergeUserInfo(userInfo)

            list.add((elem.order_detail_id,elem))
          } else {
            //从mysql中获取该用户的信息
            statement.setInt(1, elem.user_id.toInt)
            //从mysql中查询数据
            val set: ResultSet = statement.executeQuery()

            //判断mysql中是否有该用户
            while (set.next()) {
              //从mysql中获取数据并对数据进行封装
              val id: Int = set.getInt("id")
              val login_name: String = set.getString("login_name")
              val user_level: String = set.getString("user_level")
              val date: sql.Date = set.getDate("birthday")
              val gender: String = set.getString("gender")
              //转化mysql中的日期格式
              val birthday: String = new SimpleDateFormat("yyyy-MM-dd").format(date)
              val userInfo: UserInfo = UserInfo(id.toString, login_name, user_level, birthday, gender)
              elem.mergeUserInfo(userInfo)

              list.add((elem.order_detail_id, elem))
            }
          }
        }

        jedis.close()

//        statement.close()
//        conn.close()
        list.toIterator
      }
    }
/*      .foreachRDD {
        rdd => {
          rdd.foreach(println)
        }
      }*/


    //ES写入数据
    mergeUserInfoDstream.foreachRDD {
      rdd => {
        rdd.foreachPartition {
          tierSaleDetail => {
//            println("数据：" + tierSaleDetail.size)
//            tierSaleDetail.foreach(println)

//            if (tierSaleDetail.size > 0) {
              val list: List[(String, SaleDetail)] = tierSaleDetail.toList
              //往ES中插入数据
              EsUtils.insertBulk(GmallConstants.ES_INDEX_SALE_DETAIL, list)

//              tierSaleDetail.foreach(println)
//            }
          }
        }
      }
    }




    //进行两表合并可以使用join方法，但是会存在一些问题，当数据延迟发送过来的时候会出现没收到数据
    /*    val value = orderDstream
          .join(detailDstream)
          .foreachRDD {
            rdd => {
              rdd.foreach(println)
            }
          }*/



    ssc.start()

    ssc.awaitTermination()

  }
}
